package com.atguigu.day04;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.scala.typeutils.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.Random;

/**
 * @description :
 * @autor :lyh
 */
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .keyBy(r -> r.sensorId)
                .process(new TempAlert())
                .print();

        env.execute();
    }

    public static class SensorReading {
        public String sensorId;
        public double temperature;

        public SensorReading() {
        }

        public SensorReading(String sensorId, double temperature) {
            this.sensorId = sensorId;
            this.temperature = temperature;
        }

        @Override
        public String toString() {
            return "SensorReading{" +
                    "sensorId='" + sensorId + '\'' +
                    ", temperature=" + temperature +
                    '}';
        }
    }

    public static class SensorSource implements SourceFunction<SensorReading> {
        private boolean running = true;
        public Random random = new Random();


        @Override
        public void run(SourceContext<SensorReading> ctx) throws Exception {
            while (running) {
                for (int i = 1; i < 4; i++) {
                    ctx.collect(new SensorReading(
                            "sensor_" + i,
                            random.nextGaussian()
                    ));
                }
                Thread.sleep(300L);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }

    // 第一种情况：温度上升 && 不存在报警定时器
    // 1. 注册报警定时器
    // 2. 将报警定时器时间戳保存下来
    // 第二种情况：温度下降 && 存在报警定时器
    // 1. 根据报警定时器的时间戳将报警定时器从定时器队列中删除
    // 2. 删除保存的报警定时器的时间戳
    public static class TempAlert extends KeyedProcessFunction<String, SensorReading, String> {

        // 保存上一次的温度
        private ValueState<Double> lastTemp;
        // 保存报警定时器时间戳
        private ValueState<Long> timerTs;

        @Override
        public void open(Configuration parameters) throws Exception {
            lastTemp = getRuntimeContext().getState(
                    new ValueStateDescriptor<Double>(
                            "last-temp",
                            Types.DOUBLE()
                    )
            );
            timerTs = getRuntimeContext().getState(
                 new ValueStateDescriptor<Long>(
                         "timer-rs",
                         Types.LONG()
                 )
            );
        }

        @Override
        public void processElement(SensorReading in, KeyedProcessFunction<String, SensorReading, String>.Context ctx, Collector<String> out) throws Exception {
            // 取出上一次温度值
            // 当第一条温度到达时，lastTemp.value()为null
            Double prevTemp = lastTemp.value();
            // 将当前温度保存
            lastTemp.update(in.temperature);

            if (prevTemp != null) {
                // 第一种情况
                if (in.temperature > prevTemp && timerTs.value() == null) {
                    long currentProcessingTime = ctx.timerService().currentProcessingTime();
                    long oneSecondLater = currentProcessingTime + 1000L;
                    ctx.timerService().registerProcessingTimeTimer(oneSecondLater);
                    timerTs.update(oneSecondLater);
                }
                // 第二种情况
                else if(in.temperature < prevTemp && timerTs.value() != null){
                    // 删除报警定时器
                    ctx.timerService().deleteProcessingTimeTimer(timerTs.value());
                    // 由于定时器已经被手动删除，所以保存报警定时器的时间戳也要删除
                    timerTs.clear();
                }
            }
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<String, SensorReading, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            out.collect(ctx.getCurrentKey() + "连续1s温度上升");
            // onTimer 触发后，报警定时器会被自动删除，那么保存报警定时器时间戳的timerTs也需要删除
            timerTs.clear();
        }
    }
}
